package com.ling.rocketmq.listener;


import com.google.gson.Gson;
import com.google.gson.JsonArray;
import com.google.gson.JsonElement;
import com.google.gson.JsonParser;
import com.google.gson.reflect.TypeToken;
import com.ling.rocketmq.model.ConsumerMethodModel;
import com.ling.rocketmq.model.MessageBodyParam;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.ConcurrentMap;

@Component
public class LingMQMessageListener implements ApplicationContextAware, MessageListenerConcurrently {
    private final static Logger log = LoggerFactory.getLogger(LingMQMessageListener.class);
    private ApplicationContext applicationContext;
    private ConcurrentMap<String, ConsumerMethodModel> listenerConcurrentMap;

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        MessageExt msg = msgs.get(0);
        String topic = msg.getTopic();
        try {
            String message = new String(msg.getBody());
            log.info("message {}", message);
            ConsumerMethodModel listenerMethod = listenerConcurrentMap.get(topic);

            Object bean = applicationContext.getBean(listenerMethod.targetClass);

            //拿到数组
            JsonArray jsonArray = new JsonParser().parse(message).getAsJsonArray();
            Class<?>[] parameterTypes = new Class[jsonArray.size()];
            Object[] args = new Object[jsonArray.size()];

            int index = 0;

            for (JsonElement json : jsonArray) {

                MessageBodyParam param = new Gson()
                        .fromJson(
                                json,
                                new TypeToken<MessageBodyParam>() {
                                }.getType()
                        );

                parameterTypes[index] =
                        ClassLoader.getSystemClassLoader().loadClass(param.getClassName());

                args[index] =
                        new Gson().fromJson(param.getClassValue(), parameterTypes[index]);

                ++index;
            }

            bean.getClass()
                    .getMethod(listenerMethod.method.getName(), parameterTypes)
                    .invoke(bean, args);

        } catch (Exception e) {
            log.warn("消费失败，准备重试", e);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    public void setListenerConcurrentMap(ConcurrentMap<String, ConsumerMethodModel> listenerConcurrentMap) {
        this.listenerConcurrentMap = listenerConcurrentMap;
    }
}